package com.feixiang.springbootinit.processor;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import com.feixiang.springbootinit.common.ErrorCode;
import com.feixiang.springbootinit.constant.WxConstant;
import com.feixiang.springbootinit.exception.BusinessException;
import com.feixiang.springbootinit.model.entity.wechat.ChatData;
import com.feixiang.springbootinit.model.entity.wechat.QwMessage;
import com.feixiang.springbootinit.model.enums.SessionArchiveMsgActionEnum;
import com.feixiang.springbootinit.model.enums.SessionArchiveMsgTypeEnum;
import com.feixiang.springbootinit.model.vo.wechat.MsgVO;
import com.feixiang.springbootinit.service.QwMessageService;
import com.feixiang.springbootinit.strategy.MessageProcessingStrategy;
import com.feixiang.springbootinit.strategy.MessageStrategyFactory;
import com.feixiang.springbootinit.utils.SessionArchiveUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.security.PrivateKey;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Component
/**
 * 会话存档批量处理
 * 1. 从数据库中批量查询消息
 * 2. 异步处理消息
 * 3. 批量插入处理结果
 * 4. 处理完成后，更新消息状态
 * 5. 处理异常情况
 */
public class SessionArchiveBatchProcessor {
    // 线程池配置
    private static final int IO_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2; // IO密集型任务
    private static final int CPU_POOL_SIZE = Runtime.getRuntime().availableProcessors(); // CPU密集型任务
    private ExecutorService ioExecutor;
    private ExecutorService cpuExecutor;
    // 批量处理配置
    private static final int BATCH_SIZE = 500; // 每批处理的消息数量
    private static final int MAX_FUTURE_WAIT_SECONDS = 300; // 等待异步任务的最大时间
    // SDK实例（使用AtomicLong确保原子性）
    private static final AtomicLong sdkInstance = new AtomicLong(0);
    private static PrivateKey privateKey;
    private static final Object initLock = new Object();

    @Autowired
    private TransactionTemplate transactionTemplate; // 注入事务模板

    @Autowired
    private QwMessageService messageService;

    @Autowired
    private MessageStrategyFactory messageStrategyFactory;

    /**
     * 获取全局SDK实例（线程安全）
     */
    public static long getGlobalSdkInstance() {
        long instance = sdkInstance.get();
        if (instance == 0) {
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "企微会话存档SDK未初始化");
        }
        return instance;
    }

    @PostConstruct
    public void init() {
        // 初始化线程池，使用更合理的配置
        ioExecutor = new ThreadPoolExecutor(
                IO_POOL_SIZE,
                IO_POOL_SIZE,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new NamedThreadFactory("wx-io-pool"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        cpuExecutor = new ThreadPoolExecutor(
                CPU_POOL_SIZE,
                CPU_POOL_SIZE,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new NamedThreadFactory("wx-cpu-pool"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // 确保SDK只被初始化一次
        if (sdkInstance.get() == 0) {
            synchronized (initLock) {
                if (sdkInstance.get() == 0) {
                    try {
                        long instance = SessionArchiveUtil.initSDK(WxConstant.CORP_ID, WxConstant.SECRET_KEY);
                        sdkInstance.set(instance);
                        privateKey = SessionArchiveUtil.getPrivateKey(WxConstant.PRIVKEY);
                        if (privateKey == null) {
                            sdkInstance.set(0);
                            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "企微解密私钥初始化失败");
                        }
                        log.info("SessionArchiveProcessor初始化成功，SDK实例: {}", instance);
                    } catch (Exception e) {
                        log.error("SessionArchiveProcessor初始化失败", e);
                        sdkInstance.set(0);
                        throw new BusinessException(ErrorCode.SYSTEM_ERROR, "会话存档处理器初始化失败");
                    }
                }
            }
        }
    }

    @PreDestroy
    public void destroy() {
        // 关闭线程池
        shutdownExecutor(ioExecutor, "IO线程池");
        shutdownExecutor(cpuExecutor, "CPU线程池");

        // 释放SDK资源
        long instance = sdkInstance.getAndSet(0);
        if (instance != 0) {
            SessionArchiveUtil.destroy();
            log.info("SessionArchiveBatchProcessor资源已释放");
        }
    }

    /**
     * 处理消息列表
     */
    public List<QwMessage> processMessages(List<ChatData> chatDataList) {
        if (chatDataList == null || chatDataList.isEmpty()) {
            log.info("拉取的消息列表为空，无需处理");
            return Collections.emptyList();
        }

        log.info("开始处理消息，总数量: {}", chatDataList.size());

        // 并行处理消息（解密和转换）
        List<CompletableFuture<QwMessage>> futures = new ArrayList<>(chatDataList.size());
        for (ChatData chatData : chatDataList) {
            CompletableFuture<QwMessage> future = CompletableFuture
                    .supplyAsync(() -> decryptMessage(chatData), ioExecutor)
                    .thenApplyAsync(msgVO -> convertToQwMessage(msgVO, chatData), cpuExecutor);
            futures.add(future);
        }

        // 收集处理结果
        List<QwMessage> processedMessages = new ArrayList<>();
        for (CompletableFuture<QwMessage> future : futures) {
            try {
                QwMessage message = future.get(MAX_FUTURE_WAIT_SECONDS, TimeUnit.SECONDS);
                if (message != null) {
                    processedMessages.add(message);
                }
            } catch (TimeoutException e) {
                int index = futures.indexOf(future);
                ChatData chatData = chatDataList.get(index);
                log.error("消息处理超时，seq: {}", chatData.getSeq(), e);
            } catch (Exception e) {
                log.error("消息处理异常", e);
            }
        }

        // 批量保存处理结果
        if (!processedMessages.isEmpty()) {
            batchSaveMessages(processedMessages);
            log.info("消息处理完成，成功保存: {}", processedMessages.size());
        } else {
            log.info("所有消息处理失败，无数据保存");
        }

        return processedMessages;
    }

    /**
     * 批量保存消息及相关数据
     */
    private void batchSaveMessages(List<QwMessage> messages) {
        // 分批次保存，避免一次性处理过多数据
        for (int i = 0; i < messages.size(); i += BATCH_SIZE) {
            int end = Math.min(i + BATCH_SIZE, messages.size());
            List<QwMessage> batch = messages.subList(i, end);
            log.info("批量保存消息，批次: {}-{}, 数量: {}", i, end-1, batch.size());
            // 使用事务确保批次数据一致性
            int finalI = i;
            transactionTemplate.execute(status -> {
                try {
                    // 批量保存主消息
                    messageService.insertBatch(batch);
                    // 批量处理策略相关数据
                    processMessageStrategiesInBatch(batch);
                    return true;
                } catch (Exception e) {
                    status.setRollbackOnly();
                    log.error("批量保存消息异常，批次: {}-{}", finalI, end-1, e);
                    return false;
                }
            });
        }
    }
    /**
     * 批量处理消息策略
     */
    private void processMessageStrategiesInBatch(List<QwMessage> batch) {
        // 按消息类型分组处理，减少策略切换开销
        Map<String, List<QwMessage>> messagesByType = batch.stream()
                .collect(HashMap::new, (map, msg) -> {
                    String type = msg.getMsgtype();
                    map.computeIfAbsent(type, k -> new ArrayList<>()).add(msg);
                }, HashMap::putAll);

        for (Map.Entry<String, List<QwMessage>> entry : messagesByType.entrySet()) {
            String msgType = entry.getKey();
            List<QwMessage> typeMessages = entry.getValue();

            MessageProcessingStrategy processor = messageStrategyFactory.getStrategy(
                    SessionArchiveMsgTypeEnum.fromString(msgType));

            if (processor != null) {
                try {
                    processor.batchProcess(typeMessages);
                } catch (Exception e) {
                    log.error("批量处理消息策略异常, 消息类型: {}", msgType, e);
                }
            } else {
                log.warn("未找到消息类型[{}]对应的处理器，跳过处理", msgType);
            }
        }
    }

    /**
     * 解密消息
     */
    private MsgVO decryptMessage(ChatData chatData) {
        try {
            log.info("解密消息线程: {}", Thread.currentThread().getName());
            String randomKey = SessionArchiveUtil.decrptyRandomKey(privateKey, chatData.getEncryptRandomKey());
            // 使用原子方式获取SDK实例
            String decryptMsg = SessionArchiveUtil.decryptData(sdkInstance.get(), randomKey, chatData.getEncryptChatMsg());
            return JSONUtil.toBean(decryptMsg, MsgVO.class);
        } catch (Exception e) {
            log.error("消息解密失败, seq: {}", chatData.getSeq(), e);
            throw new BusinessException(ErrorCode.OPERATION_ERROR, "消息解密失败");
        }
    }

    /**
     * 转换为QwMessage对象，不包含数据库操作
     */
    private QwMessage convertToQwMessage(MsgVO msgVO, ChatData chatData) {
        try {
            log.info("转换消息线程: {}", Thread.currentThread().getName());
            return buildBaseMessage(chatData, msgVO, JSONUtil.toJsonStr(msgVO));
        } catch (Exception e) {
            log.error("构建消息对象失败, seq: {}", chatData.getSeq(), e);
            return null;
        }
    }

    private void shutdownExecutor(ExecutorService executor, String poolName) {
        try {
            executor.shutdown();
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                log.warn("{}强制关闭", poolName);
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    private QwMessage buildBaseMessage(ChatData chatData, MsgVO msgVO, String decryptMsg) {
        // 1. 参数校验
        if (chatData.getSeq() == null || StringUtils.isAnyBlank(msgVO.getMsgid(), msgVO.getAction(), decryptMsg)) {
            throw new BusinessException(ErrorCode.PARAMS_ERROR, "参数缺失: chatData.seq, msgVO.msgid, msgVO.action, decryptMsg 不能为空");
        }
        // 2. 初始化基础对象
        QwMessage message = new QwMessage();
        message.setSeq(chatData.getSeq());
        message.setMsgId(msgVO.getMsgid());
        message.setAction(msgVO.getAction());
        message.setContent(decryptMsg);
        // 3. 根据action类型处理不同分支
        String action = msgVO.getAction();
        if (SessionArchiveMsgActionEnum.SWITCH.getCode().equals(action)) {
            handleSwitchMessage(message, msgVO);
        } else if (SessionArchiveMsgActionEnum.SEND.getCode().equals(action)) {
            handleSendMessage(message, msgVO);
        } else if (SessionArchiveMsgActionEnum.RECALL.getCode().equals(action)) {
            handleSendMessage(message, msgVO); // RECALL和SEND处理逻辑相同
        } else {
            log.warn("未知的action类型: {}", action);
            throw new BusinessException(ErrorCode.PARAMS_ERROR, "非法的action类型: " + action);
        }

        return message;
    }

    // 处理切换企业日志
    private void handleSwitchMessage(QwMessage message, MsgVO msgVO) {
        message.setMsgtime(DateUtil.date(msgVO.getTime()));
        message.setMsgtype(SessionArchiveMsgActionEnum.SWITCH.getCode());
        log.debug("构建企业切换消息: msgId={}", msgVO.getMsgid());
    }

    // 处理发送/撤回消息(SEND和RECALL共用)
    private void handleSendMessage(QwMessage message, MsgVO msgVO) {
        message.setFromId(msgVO.getFrom());
        message.setMsgtype(msgVO.getMsgtype());
        message.setTolist(JSONUtil.toJsonStr(msgVO.getTolist()));
        message.setRoomid(msgVO.getRoomid());
        Date msgTime = DateUtil.date(msgVO.getMsgtime());
        message.setMsgtime(msgTime);
        message.setCreatedTime(new Date());
        log.debug("构建{}消息: msgId={}, 发送时间={}",
                message.getAction(), message.getMsgId(), msgTime);
    }

    /**
     * 线程工厂
     */
    static class NamedThreadFactory implements ThreadFactory {
        private static final Logger log = LoggerFactory.getLogger(NamedThreadFactory.class);
        private final String namePrefix;
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        NamedThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
        }

        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, namePrefix + "-" + threadNumber.getAndIncrement());
            // 设置为守护线程，避免影响应用关闭
            thread.setDaemon(true);
            // 设置线程优先级为普通
            thread.setPriority(Thread.NORM_PRIORITY);
            log.debug("创建线程: {}", thread.getName());
            return thread;
        }
    }
}